package com.gzyj.flink.gps.map.area;

import cn.hutool.core.lang.Pair;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.AreaData;
import com.gzyj.flink.dto.alarm.RemindContent;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.mapper.mysql.MapVehareaMapper;
import com.gzyj.flink.gps.mapper.mysql.MapareaMapper;
import com.gzyj.flink.gps.service.AreaService;
import com.gzyj.flink.gps.service.NormalAlarmService;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.mysql.set.Maparea;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.*;
import java.util.concurrent.TimeUnit;

public class AreaAlarmDataProcess extends RichFlatMapFunction<GPSPosition, AreaAlarmResult> {
    LettuceConnectionFactory redisFactory;

    SqlSessionFactory mysql;

    private AreaService areaService;

    private NormalAlarmService normalAlarmService;

    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        redisFactory = RedisSessionFactory.getRedisFactory();
        mysql = MySQLMybatisSessionFactory.getSqlSessionFactory();
        areaService = new AreaService();
        normalAlarmService = new NormalAlarmService();
    }

    public Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
        if (obj == null) {
            return new Driverlogininfo();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
        }

    }

    @Override
    public void flatMap(GPSPosition gpsPosition, Collector<AreaAlarmResult> collector) throws Exception {
        RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
        SqlSession sqlSession = mysql.openSession(true);

        try {

            MapVehareaMapper mapVehareaMapper = sqlSession.getMapper(MapVehareaMapper.class);
            MapareaMapper mapareaMapper = sqlSession.getMapper(MapareaMapper.class);


            Driverlogininfo driverLoginData = TryGetData_DLD(gpsPosition.getVehId(), clusterRedisTemplate);

            Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_Area + gpsPosition.getVehId());
            //获取运行数据
            AreaData areaData;
            if (obj == null) {
                areaData = new AreaData();
            } else {
                areaData = JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), AreaData.class);
            }

            Pair<Boolean, Integer> p = TimeUtils.CheckData(areaData.getGpsTime(), gpsPosition.getTIME());
            int i = p.getValue();
            if (!p.getKey()) {
                return;
            } else {
                areaData.GpsTime = gpsPosition.getGPSTimeExt();
            }

            boolean isInArea = areaData.InArea;
            Maparea mySetting = areaService.InArea(gpsPosition.getVehId(), gpsPosition.getLngExt(),
                    gpsPosition.getLatExt(), false, mapVehareaMapper, mapareaMapper);


            if (mySetting != null && !isInArea) {
                if (mySetting == null) return;
                //这次在区域内，但上次没在区域，则报进区域提醒
                areaData.InArea = true;
                areaData.AreaId = mySetting.getAreaid();
                areaData.AreaName = mySetting.getName();
                areaData.KeepTimes = 0;
                areaData.InTime = gpsPosition.getGPSTimeExt();
                areaData.AlarmGuid = UUID.randomUUID().toString().replaceAll("-", "");
                areaData.Longitude = gpsPosition.getLngExt();
                areaData.Latitude = gpsPosition.getLatExt();
                clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_Area + gpsPosition.getVehId(), areaData, 8, TimeUnit.HOURS);
                AreaAlarmResult areaAlarmResult = new AreaAlarmResult();
                areaAlarmResult.setKafkaVehicleAreaAlarm(normalAlarmService.SendAlarm(areaData, gpsPosition, driverLoginData));
                if (StringUtils.isNotEmpty(mySetting.getTxnr())) {
                    GPSPosition finalGpsPosition = gpsPosition;
                    RemindContent remindContent = new RemindContent() {
                        {
                            VehicleId = gpsPosition.getVehId();
                            Ipaddress = gpsPosition.getTerminalPhoneNo();
                            VehicleNo = gpsPosition.getCheliangpaizhao();
                            PlateColor = gpsPosition.getChepaiyanse();
                            RemindNote = mySetting.getTxnr();
                            RemindType = 2;
                            RemindTime = new Date();
                            MessageType = "进区域报警";
                            AlarmNote = "进区域报警";
                            GpsTime = areaData.GpsTime;
                            LastSpeed = finalGpsPosition.getSpeedExt();
                        }

                    };
                    areaAlarmResult.setKafkaVehicleRemindContent(normalAlarmService.SendRemindToKafka(remindContent));
                }

                collector.collect(areaAlarmResult);
            } else if (mySetting != null && isInArea) {
                if (mySetting == null) return;
                //这次在区域内，上次也在区域,比较2次区域有无差别
                if (areaData.AreaId != mySetting.getAreaid()) {
                    //这次进的区域和上次进的区域不一样
                    //则报一个出区域报警
                    areaData.InArea = false;
                    //areaData.AreaId = -1; 保留进区域的数据
                    //areaData.AreaName = ""; 保留进区域的数据
                    areaData.KeepTimes = areaData.InTime == TimeUtils.MIN_VALUE() ? 0 : TimeUtils.getMinusSecond(gpsPosition.getGPSTimeExt(), areaData.InTime);
                    areaData.InTime = TimeUtils.MIN_VALUE();
                    //areaData.AlarmGuid = Guid.NewGuid().ToString("N");保留进区域的数据
                    areaData.Longitude = gpsPosition.getLngExt();
                    areaData.Latitude = gpsPosition.getLatExt();
                    //存
                    clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_Area + gpsPosition.getVehId(), areaData, 8, TimeUnit.HOURS);

                    collector.collect(new AreaAlarmResult(normalAlarmService.SendAlarm(areaData, gpsPosition, driverLoginData), null));
                    //然后报一个进区域报警
                    areaData.InArea = true;
                    areaData.AreaId = mySetting.getAreaid();
                    areaData.AreaName = mySetting.getName();
                    areaData.KeepTimes = 0;
                    areaData.InTime = gpsPosition.getGPSTimeExt();
                    areaData.AlarmGuid = UUID.randomUUID().toString().replaceAll("-", "");
                    areaData.Longitude = gpsPosition.getLngExt();
                    areaData.Latitude = gpsPosition.getLatExt();

                    clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_Area + gpsPosition.getVehId(), areaData, 8, TimeUnit.HOURS);
                    collector.collect(new AreaAlarmResult(normalAlarmService.SendAlarm(areaData, gpsPosition, driverLoginData), null));

                    if (StringUtils.isNotEmpty(mySetting.getTxnr())) {
                        GPSPosition finalGpsPosition1 = gpsPosition;
                        RemindContent remindContent = new RemindContent() {
                            {
                                VehicleId = gpsPosition.getVehId();
                                Ipaddress = gpsPosition.getTerminalPhoneNo();
                                VehicleNo = gpsPosition.getCheliangpaizhao();
                                PlateColor = gpsPosition.getChepaiyanse();
                                RemindNote = mySetting.getTxnr();
                                RemindType = 2;
                                RemindTime = new Date();
                                MessageType = "进区域报警";
                                AlarmNote = "进区域报警";
                                GpsTime = areaData.GpsTime;
                                LastSpeed = finalGpsPosition1.getSpeedExt();
                            }

                        };
                        collector.collect(new AreaAlarmResult(null, normalAlarmService.SendRemindToKafka(remindContent)));
                    }


                } else {
                    clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_Area + gpsPosition.getVehId(), areaData, 8, TimeUnit.HOURS);
                }
            } else if (mySetting == null && !isInArea) {

                //这次没在区域内，上次也没在区域，则存储数据
                //无需发送出区域，本身就没在区域里,不存在出区域
                areaData.InArea = false;
                areaData.AreaId = -1;
                areaData.AreaName = "";
                areaData.KeepTimes = 0;
                areaData.InTime = TimeUtils.MIN_VALUE();
                areaData.AlarmGuid = UUID.randomUUID().toString().replaceAll("-", "");
                areaData.Longitude = 0;
                areaData.Latitude = 0;

                clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_Area + gpsPosition.getVehId(), areaData, 8, TimeUnit.HOURS);

            } else if (mySetting == null && isInArea) {
                //这次没在区域内，上次在区域，则报出区域提醒
                areaData.InArea = false;
                //areaData.AreaId = -1; 保留进区域的数据
                //areaData.AreaName = ""; 保留进区域的数据
                areaData.KeepTimes = areaData.InTime == TimeUtils.MIN_VALUE() ? 0 : TimeUtils.getMinusSecond(gpsPosition.getGPSTimeExt(), areaData.InTime);
                areaData.InTime = TimeUtils.MIN_VALUE();
                //areaData.AlarmGuid = Guid.NewGuid().ToString("N");保留进区域的数据
                areaData.Longitude = gpsPosition.getLngExt();
                areaData.Latitude = gpsPosition.getLatExt();

                clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_Area + gpsPosition.getVehId(), areaData, 8, TimeUnit.HOURS);

                collector.collect(new AreaAlarmResult(normalAlarmService.SendAlarm(areaData, gpsPosition, driverLoginData), null));
            }


        } catch (Exception e) {
            e.printStackTrace();
            sqlSession.rollback();
        } finally {
            sqlSession.close();
        }

    }
}
